package com.aotain.project.apollo.bolt;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 端口异常统计
 * 按照目的IP，统计各个异常端口的访问次数
 * @author Administrator
 *
 */
public class PortAbnStatisBolt extends AbstractBolt{

	
	HashMap<String,Long> _mapPortStat = new HashMap<String,Long>();

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println("PortAbnStatisBolt exception "+e.getMessage());
			Logger.getRootLogger().error("PortAbnStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}
	
	private void countInLocal(Tuple tuple) {
		String dip = tuple.getStringByField("dip");
		String dport = tuple.getStringByField("dport");
		String dateStr_d = tuple.getStringByField("dateStr_d"); 
		
		long triplePV = tuple.getLongByField("pv");
		
		
		String rowKey = String.format("%s_%s_%s", dip, dateStr_d, dport);
		if(_mapPortStat.containsKey(rowKey))
		{
			_mapPortStat.put(rowKey, _mapPortStat.get(rowKey) + triplePV);
		}
		else
		{
			_mapPortStat.put(rowKey, triplePV);
		}
	}
	
	private void emitCountingData(OutputCollector collector) {
		
		for(Entry<String,Long> entry : _mapPortStat.entrySet()) {
			
			String rowKey = entry.getKey();
			long count = entry.getValue();
			
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			
		
			String dip = entry.getKey().split("_",-1)[0];
			String strDateD = entry.getKey().split("_",-1)[1];
			String dPort = entry.getKey().split("_",-1)[2];
			
			//String rowkey = String.format("%s_%s_%s", strDateD, dip, port);
			
			String tbName = ApolloProperties.SDS_PORT_STAT_DAY;
			String cf = "cf";
			
			
			hbaseInstance.Add(tbName, rowKey, cf,  "REPORTDATE", strDateD);
			hbaseInstance.Add(tbName, rowKey, cf,  "IP", dip);
			hbaseInstance.Add(tbName, rowKey, cf,  "PORT", dPort);
			hbaseInstance.incerment(tbName, rowKey, "cf:PORTNUM", count);
		}
		
		_mapPortStat.clear();
		
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,900);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
